/*
This application starts two threads, one for sending and one for receiving of messages over the network. Both send and receive work in a mechanism whereby they put messages in queues before sending or after receiving the messages.
*/
#include "talk.h"

pthread_mutex_t mute1 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mute2 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex4 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex5 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex3 = PTHREAD_MUTEX_INITIALIZER;

const int size_type = ( sizeof(int)/sizeof(char) );	//Tells the size of the character array storing the type of message being exchanged.
/*
This function starts exchanging messages between the queues and the network.
The argument node_no is essentially char* form of the index which has the computer's address (Argument passed as node_no should be node[i] if the ip address of the machine which runs this is in add[i]).
*/
void *main_sendrecv(void *node_no)
{
  char *node_no_ch;
  int ret1, ret2;
  pthread_t sendthread, recvthread;

  if((sockfd = socket(AF_INET, SOCK_DGRAM, 0))==-1)
    {
      perror("socket");
      exit(1);
    }
  queue_init();
  node_no_ch = (char *)node_no;
  hno = server_check(node_no_ch);	//Save the index corresponding to this node in hno
  pthread_mutex_lock(&mutex3);
  add_init((int)MYPORT);		//Initialize addresses using MYPORT as port
  pthread_mutex_unlock(&mutex3);
  if(bind(sockfd, (struct sockaddr *)&addr[hno], sizeof(struct sockaddr))==-1)
    {
      perror("bind");
      exit(1);
    }
  if((ret1 = pthread_create(&sendthread, NULL, &fnsend, NULL)))		//Thread for sending onto network, runs parallel to other threads
   {
      	printf("Thread creation failed: %d\n", ret1);
   }
  if((ret2=pthread_create( &recvthread, NULL, &fnrecv, NULL)))		//Thread for receiving from network, also runs parallel to other threads
   {
        printf("Thread creation failed: %d\n", ret2);
   }
//The following two statements ensure that the function ends only after both sendthread and recvthread return, or join this thread. For our case, it never happens and hence it runs for ever.
   pthread_join(sendthread, NULL);
   pthread_join(recvthread, NULL);					
   return (NULL);
}

//Function to initialize send and receive queues
void queue_init()
{
  sendQ = NULL;
  recvQ = NULL;
}

//This function sends all elements which have been enqueued in the queue, sendQ onto the network according to FIFO. It runs forever.
void *fnsend()
{
  struct sockaddr_in their_addr;
  char *m;
  int i, size;
  sn_s *tmpsQ1, *tmpsQ2;			//Variables used to access appropriate element of the queue
  tmpsQ1 = tmpsQ2 = NULL;
  while(1)
    {
      if(sendQ!=NULL)                           //If queue is not empty
	{
	  pthread_mutex_lock(&mute1);
	  tmpsQ1 = sendQ;
	  tmpsQ2 = tmpsQ1->next;
	  if(tmpsQ2!=NULL)                 //If queue has atleast two elements
	    {
	      while(tmpsQ2->next!=NULL)		//Go to last and second last elements
		{
		  tmpsQ1 = tmpsQ1->next;
		  tmpsQ2 = tmpsQ2->next;
		}
	      size = tmpsQ2->size_m;
	      their_addr = tmpsQ2->dest_add;
	      m = (char *)malloc((size + 1)*sizeof(char));
	      for(i=0;i<size;i++)
		*(m + i) = *(tmpsQ2->m + i);
	      free(tmpsQ2->m);				
	      *(m + size) = '\0';			//As message is sent in char* form, this denotes the end of string
	      if(sendto(sockfd, m, size * sizeof(char), 0, (struct sockaddr *) &their_addr, sizeof(struct sockaddr))==-1)
		{
		  perror("sendto");
		  exit(1);
		}
	      //else
		//printf("Send successful: %d\n", tmpsQ2->type);
	      free(m);
	      tmpsQ1->next = NULL;
	      free(tmpsQ2);
	    }
	  else              				//If queue has a single element
	    {
	      size = tmpsQ1->size_m;
	      their_addr = tmpsQ1->dest_add;
	      m = (char *)malloc((size + 1)*sizeof(char));
	      for(i=0;i<size;i++)
		*(m + i) = *(tmpsQ1->m + i);
	      free(tmpsQ1->m);				//Freeing the message from the queue
	      *(m + size) = '\0';			//As message is sent in char *form, this denotes the end of string
	      if(sendto(sockfd, m, size * sizeof(char), 0, (struct sockaddr *) &their_addr, sizeof(struct sockaddr))==-1)
		{
		  perror("sendto");
		  exit(1);
		}
	      //else
		//printf("Send successful: %d\n", *((int*)m));
	      free(m);
	      sendQ=NULL;
	      free(tmpsQ1);
	    }
	  pthread_mutex_unlock(&mute1);
	}
    }
  return(NULL);
}

//Function to enque message msg into sendQ. size_msg has the size of the actual message in bytes. Call this function from your application to enqueue.
void queue_send(void *msg, size_t size_msg, struct sockaddr_in dest_addr)
{
  sn_s *tmps;
  char *tmp_ptr, typ_char[size_type + 1];
  size_t size_ch = sizeof(char);
  int i, size, *typ_ptr;
  tmps=(sn_s *)malloc(sizeof(sn_s));			//Creating a new node in the queue
  tmp_ptr = (char *)msg;
  tmps->m=(char *)malloc(size_msg + sizeof(char));		//New message created
  size = (int)(size_msg/size_ch);
  for(i=0;i<size;i++)				//Existing message copied into queue
    *(tmps->m + i) = *(tmp_ptr + i);
  *(tmps->m + size)='\0';

  for(i=0;i<size_type;i++)		//Copying the type of message to a different variable
    typ_char[i] = *(tmp_ptr + i);
  typ_char[size_type]='\0';

  typ_ptr = (int *)typ_char;			//Type of message being typecast to int form
//Message being assigned appropriate parameters.
  tmps->type = *typ_ptr;
  tmps->size_m = size;
  tmps->dest_add = dest_addr;
  pthread_mutex_lock(&mutex5);
  tmps->next = sendQ;			//Enqueueing message in front of the linked list
  sendQ=tmps;
  tmp_ptr = NULL;
  pthread_mutex_unlock(&mutex5);
}

/*
This functions receives from network and puts data into the queue recvQ. It runs forever.
*/
void *fnrecv()
{
  char *tmp_msg, typ_char[size_type + 1];
  unsigned add_len;
  int numbytes, i;
  int *type_ptr;
  sn_r *tmpr;
  tmp_msg = (char *)malloc(MAXBUFFLEN * sizeof(char));			//Message is received in string form. The maximum possible size of the message is denoted by MAXBUFFLEN (defined in config)
  struct sockaddr_in their_addr;
  while(1)
    {
      add_len = sizeof(struct sockaddr);
      if((numbytes=recvfrom(sockfd, tmp_msg, (MAXBUFFLEN-1)*sizeof(char), 0, (struct sockaddr *) &their_addr, &add_len)) == -1)
	{
	  perror("recvfrom");
	  continue;
	}
      else if(numbytes>0)
	{
	  tmpr = (sn_r *)malloc(sizeof(sn_r));                      //New Node in the recvQ is introduced at the begining
      	  tmpr->msw = (char *)malloc((numbytes+1)*sizeof(char));
	  for(i=0;i<numbytes;i++)
	    *(tmpr->msw + i) = *(tmp_msg + i);
	  *(tmpr->msw + numbytes) = '\0';
	  for(i=0;i<size_type;i++)
	    typ_char[i] = *(tmp_msg + i);
	  typ_char[size_type]='\0';
	  type_ptr = (int *)typ_char;
	  tmpr->type = *type_ptr;
	  tmpr->size_m = numbytes;
	  tmpr->source_addr = their_addr;
	  pthread_mutex_lock(&mute2);
	  tmpr->next = recvQ;
	  recvQ = tmpr;
	  pthread_mutex_unlock(&mute2);
	  type_ptr = NULL;
	  //printf("Received & queued successfully\n");
	}
      //else
	//sleep(1);
    }
  return (NULL);
}

/*
Dequeueing a message of the type typ, if it exists and returning it as a type msg_s.
The ip address of the source is stored in source_addr and size of the message is stored in size_msg
*/
int queue_recv(char *mw, int typ, int *size_msg, struct sockaddr_in *source_addr)
{
  sn_r *tmprQ1, *tmprQ2, *newQ, *tmpnQ1;
  tmprQ1 = tmprQ2 = newQ = tmpnQ1 = NULL;
  int *typ_ptr, flag = 0, i, size;      //flag says if message of desired type is present or not.
  if(recvQ!=NULL)              //If queue is not empty
    {
      pthread_mutex_lock(&mutex4);
      while(recvQ!=NULL)
	{
	  tmprQ1 = recvQ;
	  if(tmprQ1->next!=NULL)                //If queue has atleast two elements
	    {
	      tmprQ2=tmprQ1->next;
	      while(tmprQ2->next!=NULL)
		{
		  tmprQ1=tmprQ1->next;
		  tmprQ2=tmprQ2->next;
		}
	      typ_ptr = (int *)tmprQ2;
	      if(*typ_ptr == typ)		//If message is of the required type then it is sent to the application.
		{
		  size = tmprQ2->size_m;
		  for(i=0;i<=size;i++)		//Creating a copy of the message into the argument sent
		    *(mw + i) = *(tmprQ2->msw + i);
		  *size_msg = size;			//Returning the size of the message
		  free(tmprQ2->msw);			
		  *source_addr = tmprQ2->source_addr;  //Returning the address from which message was received
		  free(tmprQ2);
		  tmprQ1->next = NULL;
		  flag = 1;
		  break;
		}
	      else			//Make a new queue, newQ which stores all messages which are not of the desired type.
		{
		  tmprQ2->next = newQ;		//Enqueue node tmprQ2 to the begining of newQ
		  newQ = tmprQ2;
		  tmprQ1->next = NULL;
		}
	    }
	  else                                  //If queue has single element
	    {
	      typ_ptr = (int *)tmprQ1;
	      if(*typ_ptr == typ)            //If message is of the required type then it is sent to the application
		{
		  size = tmprQ1->size_m;
      		  for(i=0;i<=size;i++)
		    *(mw + i) = *(tmprQ1->msw + i);
		  *size_msg = size;
		  free(tmprQ1->msw);
		  *source_addr = tmprQ1->source_addr;
		  free(tmprQ1);
		  recvQ = NULL;
		  flag = 1;
		  break;
		}
	      else			//Make a new queue, newQ which stores all messages which are not of the desired type.
		{
		  tmprQ1->next = newQ;
		  newQ = tmprQ1;
		  recvQ = NULL;
		}
	    }
	}
      if(newQ!=NULL)        //enqueueing the queue newQ which has messages of type!=typ back into the already existing queue recvQ
	{
		tmpnQ1 = newQ;
	  	while(tmpnQ1->next!=NULL)
	    	{
		      tmpnQ1 = tmpnQ1->next;
	        }
	  tmpnQ1->next=recvQ;
	  recvQ = newQ; 
	  newQ = NULL;
	}
      pthread_mutex_unlock(&mutex4);
      if(flag == 1)
	return 1;
    }
  return -1;
}
